package com.swhy

import java.net.URI
import java.util.{Date, UUID}

import com.utils._
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.{SparkConf, SparkContext}
import org.codehaus.jackson.map.ObjectMapper
import org.slf4j.LoggerFactory


object ChgeCustStatTotal2 {
  def main(args: Array[String]): Unit = {
    val logger = LoggerFactory.getLogger(this.getClass)

    if (args.length != 3) {
      println(
        """
          |com.swhy.CompanyInfo
          |params:
          | InputDate    刷数日期
          | OutputPath   输出路径
          | SplitNum     文件生成数量
        """.stripMargin)
      sys.exit()
    }
    val conf = new Configuration()
    val Array(inputDt, outputPath, splitNum) = args
    val sparkConf = new SparkConf().setAppName(this.getClass.getName)
    val sc = new SparkContext(sparkConf)
    val hiveContext = new HiveContext(sc)
//改下方的sql为需要装换成jason的sql
    val allcompany =s"""
        |select
        |distinct
        |mgr_cd,
        |brch_org_cd,
        |brch_org_level_cd,
        |comroppy_chg_cust_cnt,
        |risk_chg_cust_cnt,
        |chg_cust_cnt,
        |eff_cust_vol,
        |ast_sum,
        |ast_chg_sum,
        |risk_ast_chg_sum,
        |comroppy_ast_chg_sum,
        |finpd_val,
        |chg_finpd_val,
        |comroppy_finpd_val,
        |risk_finpd_val,
        |duration_defc_cnt as FX01,
        |duration_prft_cnt as SJ01,
        |l1y_new_stk_allo_vol_3 as SJ02,
        |week_logon_upcnt as SJ03,
        |week_logon_downcnt as FX02,
        |conti_1months_ast_1 as FX03,
        |bjs_pros_cnt as SJ04,
        |magr_pros_cnt as SJ05,
        |sh_pros_cnt as SJ06,
        |opt_pros_cnt as SJ07,
        |ai_pros_cnt as FX04,
        |tm,
        |biz_date
        |from
        |app_data.app_dao_ind_custchgsum_d_q a
        |where biz_date=${inputDt}
      """.stripMargin
    print(allcompany)
    val frame = hiveContext.sql(allcompany)
    val result = frame.rdd.map(x => {
      var cominfo = new StringBuilder()

      val mgr_cd = x.getAs[String]("mgr_cd")
//      val mgr_name = x.getAs[String]("mgr_name")
      val brch_org_cd = x.getAs[String]("brch_org_cd")
      val brch_org_level_cd = x.getAs[String]("brch_org_level_cd")

      val comroppy_chg_cust_cnt = x.getAs[Long]("comroppy_chg_cust_cnt")

      val risk_chg_cust_cnt = x.getAs[Long]("risk_chg_cust_cnt")

      val chg_cust_cnt = x.getAs[Long]("chg_cust_cnt")

      val eff_cust_vol = x.getAs[Long]("eff_cust_vol")


      //scala中获取decimal类型数据
      val ast_sum =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("ast_sum"))
      //转换为java的bigDecimal
      val ast_sum1  = ast_sum.bigDecimal

      val ast_chg_sum =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("ast_chg_sum"))
      val ast_chg_sum1  = ast_chg_sum.bigDecimal

      val risk_ast_chg_sum =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("risk_ast_chg_sum"))
      val risk_ast_chg_sum1  = risk_ast_chg_sum.bigDecimal

      val comroppy_ast_chg_sum =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("comroppy_ast_chg_sum"))
      val comroppy_ast_chg_sum1  = comroppy_ast_chg_sum.bigDecimal

      val finpd_val =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("finpd_val"))
      val finpd_val1  = finpd_val.bigDecimal

      val chg_finpd_val =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("chg_finpd_val"))
      val chg_finpd_val1  = chg_finpd_val.bigDecimal

      val comroppy_finpd_val =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("comroppy_finpd_val"))
      val comroppy_finpd_val1  = comroppy_finpd_val.bigDecimal

      val risk_finpd_val =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("risk_finpd_val"))
      val risk_finpd_val1  = risk_finpd_val.bigDecimal

      val duration_defc_cnt = x.getAs[Long]("FX01")

      val duration_prft_cnt = x.getAs[Long]("SJ01")

      val l1y_new_stk_allo_vol_3 = x.getAs[Long]("SJ02")

      val week_logon_upcnt = x.getAs[Long]("SJ03")

      val week_logon_downcnt = x.getAs[Long]("FX02")

      val conti_1months_ast_1 = x.getAs[Long]("FX03")

      val bjs_pros_cnt = x.getAs[Long]("SJ04")

      val magr_pros_cnt = x.getAs[Long]("SJ05")

      val sh_pros_cnt = x.getAs[Long]("SJ06")

      val opt_pros_cnt = x.getAs[Long]("SJ07")

      val ai_pros_cnt = x.getAs[Long]("FX04")


      val tm = x.getAs[Date]("tm")
      val biz_date = x.getAs[Integer]("biz_date")


      val _id = new ID()
      val index = new Index()
      val companyInfo = new ChgeCustStatTotal2()

      _id.set_id(UUID.randomUUID().toString)
      index.setIndex(_id)
      companyInfo.setMgrCd(mgr_cd)
//      companyInfo.setMgrName(mgr_name)
      companyInfo.setBrchOrgCd(brch_org_cd)
      companyInfo.setBrchOrgLevelCd(brch_org_level_cd)
      companyInfo.setComroppyChgCustCnt(comroppy_chg_cust_cnt)
      companyInfo.setRiskChgCustCnt(risk_chg_cust_cnt)
      companyInfo.setChgCustCnt(chg_cust_cnt)
      companyInfo.setEffCustVol(eff_cust_vol)
      companyInfo.setAstSum(ast_sum1)
      companyInfo.setAstChgSum(ast_chg_sum1)
      companyInfo.setRiskAstChgSum(risk_ast_chg_sum1)
      companyInfo.setComroppyAstChgSum(comroppy_ast_chg_sum1)
      companyInfo.setFinpdVal(finpd_val1)
      companyInfo.setChgFinpdVal(chg_finpd_val1)
      companyInfo.setComroppyFinpdVal(comroppy_finpd_val1)
      companyInfo.setRiskFinpdVal(risk_finpd_val1)

      companyInfo.setFX01(duration_defc_cnt)
      companyInfo.setSJ01(duration_prft_cnt)
      companyInfo.setSJ02(l1y_new_stk_allo_vol_3)
      companyInfo.setSJ03(week_logon_upcnt)
      companyInfo.setFX02(week_logon_downcnt)
      companyInfo.setFX03(conti_1months_ast_1)
      companyInfo.setSJ04(bjs_pros_cnt)
      companyInfo.setSJ05(magr_pros_cnt)
      companyInfo.setSJ06(sh_pros_cnt)
      companyInfo.setSJ07(opt_pros_cnt)
      companyInfo.setFX04(ai_pros_cnt)


      companyInfo.setTm(tm)
      companyInfo.setBizDate(biz_date)


      val mapper = new ObjectMapper()

      val indexJson = mapper.writeValueAsString(index) //企业自增编码json串
      val companyInfoJson = mapper.writeValueAsString(companyInfo) //企业信息串

      cominfo = cominfo.append(indexJson + "\n" + companyInfoJson) //json结果串
      cominfo.toString()
    })
    val path = new Path(outputPath) //创建文本输出路径
    val fs = FileSystem.get(URI.create(outputPath), conf)
    if (fs.exists(path)) {
      fs.delete(path, true)
    }
    result.coalesce(splitNum.toInt,true).saveAsTextFile(outputPath)

    sc.stop()
  }
}
